package org.niit.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Properties;
import java.util.Set;

// 指定偏移量消费数据
public class CustomConsumerSeek {

    public static void main(String[] args) throws InterruptedException {
        //1.配置信息
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092");
        //指定消费者组
        props.put(ConsumerConfig.GROUP_ID_CONFIG,"W2");
        //开启自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");
        //自动提交的时间间隔
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");
        //消费数据的类型 1.latest（默认类型，消费最新的数据） 2.earliest 消费历史数据
        //props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        //将消费出来的数据进行转换 二进制---》String
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());

        //2.创建Kafka消费者
        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(props);

        //3.订阅组主题
       consumer.subscribe(Arrays.asList("BD2") );

        //4.1获取当前主题下所有分区信息
        Set<TopicPartition> assignment = consumer.assignment();//可能获得是一个空的分区信息
        //4.2为了保证所有分区都被重置偏移量，要先判断当前 获得的分区信息的长度是否为0（空）
        //如果分区信息长度为0（空）,那就要一直获取分区信息，直到不为0(空)为止
        while (assignment.size() == 0){
            consumer.poll(Duration.ofSeconds(1));
            //将初始化后的分区信息重新给 assignment
            assignment=  consumer.assignment();
        }
        //4.3将要移动的偏移量 80 分配到每一个分区当中
        for(TopicPartition tp:assignment){
            consumer.seek(tp,80);
        }

        //5.开启死循环，不断的从Kafka中获取数据
        for(;;){

            //4.1 Kafka的消费者一次会拉去一批（很多数据）的数据
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(3));
            //4.2对这一批数据进行逐条遍历打印
            for (ConsumerRecord<String,String> data:records){
                String topic = data.topic();
                String value = data.value();
                long offset = data.offset();
                System.out.println("主题："+topic+",偏移量："+offset+",值："+value);


                Thread.sleep(1000);
            }
        }



    }
}
